package io.xxx.mario;

import io.reactivex.rxjava3.core.BackpressureStrategy;
import io.reactivex.rxjava3.core.Flowable;
import io.reactivex.rxjava3.core.FlowableOnSubscribe;
import io.reactivex.rxjava3.disposables.Disposable;
import io.xxx.mario.task.Task;
import io.xxx.mario.task.TaskMapper;
import io.xxx.mario.task.TaskService;
import org.quartz.JobExecutionException;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.transaction.annotation.Transactional;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;

import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;

@RestController
@RequestMapping("/test")
public class TestController {

    @Autowired
    private TaskMapper taskMapper;

    @Autowired
    private TaskService taskService;

    public static void main(String[] args) throws InterruptedException {
        CountDownLatch latch = new CountDownLatch(1);
        Flowable.create((FlowableOnSubscribe<Integer>) emitter -> {
                    for (int i = 0; i < 20; i++) {
                        emitter.onNext(i);
                        TimeUnit.MILLISECONDS.sleep(500);
                    }
                    emitter.onComplete();
                }, BackpressureStrategy.BUFFER)
                .doOnComplete(latch::countDown)
//                .blockingSubscribe(System.out::println);
                .subscribe(System.out::println);
        latch.await();
    }

    @GetMapping("/push")
    public void testPush() throws InterruptedException {
        Task task = taskMapper.selectById(1L);
        taskService.push(task);
    }

    @Transactional
    @GetMapping("/cursor")
    public void testCursor() throws Exception {
        Disposable disposable = Flowable.create((FlowableOnSubscribe<Integer>) emitter -> {
                    for (int i = 0; i < 10; i++) {
                        emitter.onNext(i);
                        TimeUnit.SECONDS.sleep(1);
                    }
//                    Cursor<TaskReceiver> cursor = taskReceiverMapper.findUncompletedTaskReceivers(1L);
//                    for (TaskReceiver taskReceiver : cursor) {
//                        emitter.onNext(taskReceiver);
//                    }
//                    cursor.close();
                    TimeUnit.SECONDS.sleep(5);
                    emitter.onComplete();
                }, BackpressureStrategy.BUFFER)
                .subscribe(System.out::println);

        System.out.println(disposable.isDisposed());
        TimeUnit.SECONDS.sleep(20);
        System.out.println(disposable.isDisposed());
    }
}
